package org.spbsu.apco.server.broker.impl;

import org.perf4j.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spbsu.apco.common.Pair;
import org.spbsu.apco.common.TaskService;
import org.spbsu.apco.common.monitoring.event.TaskEvent;
import org.spbsu.apco.common.properties.ApplicationPropertiesManager;
import org.spbsu.apco.common.properties.PropertyKey;
import org.spbsu.apco.common.serialization.Serializer;
import org.spbsu.apco.common.task.core.BaseTask;
import org.spbsu.apco.common.task.core.BatchTask;
import org.spbsu.apco.common.task.core.Ticket;
import org.spbsu.apco.common.utils.NetworkUtils;
import org.spbsu.apco.common.utils.StopWatchFactory;
import org.spbsu.apco.server.broker.Broker;
import org.spbsu.apco.server.scheduler.Scheduler;
import org.spbsu.apco.server.taskset.BaseTaskSet;
import org.spbsu.apco.server.taskset.TaskSetEmitter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.FileSystemXmlApplicationContext;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.*;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import java.io.File;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * User: solozobov
 * Date: 11.11.12
 */
@Path("/task")
public class BrokerImpl implements Broker, TaskService, ApplicationContextAware, InitializingBean {
    private static final Logger log = LoggerFactory.getLogger(BrokerImpl.class);
    private static final AtomicInteger instances = new AtomicInteger(0);

    private Scheduler scheduler;
    @Resource(name = "serializer")
    private Serializer serializer;
    @Context
    private HttpServletResponse servletResponse;
    private TaskSetEmitter defaultTaskSetEmitter;

    public BrokerImpl() {
        log.debug("Broker init()");
        if (instances.incrementAndGet() > 1) {
            throw new IllegalStateException("Cannot initialize more than 1 broker per JVM.");
        }
        String taskSetDescriptorFileName = ApplicationPropertiesManager.getProperty(PropertyKey.TASKSET_DESCRIPTOR_FILENAME);
        if (taskSetDescriptorFileName != null) {
            Assert.isTrue(new File(taskSetDescriptorFileName).exists(), "Taskset descriptor file not found: " + new File(taskSetDescriptorFileName).getAbsolutePath());
            final ApplicationContext context = new FileSystemXmlApplicationContext(taskSetDescriptorFileName);
            Map<String,TaskSetEmitter> taskSetEmitters = context.getBeansOfType(TaskSetEmitter.class);
            Assert.isTrue(taskSetEmitters.size()==1, "There should be exactly one TaskSetEmitter defined while there are " +
                                                        taskSetEmitters.size() + " in " + taskSetDescriptorFileName);
            this.defaultTaskSetEmitter = taskSetEmitters.values().iterator().next();
            defaultTaskSetEmitter.setBroker(this);
        }
    }


    @Override
    public void addTaskSet(final BaseTaskSet taskSet) {
        scheduler.holdTaskSet(taskSet);
    }

    @Override
    public boolean isFinished(BaseTaskSet taskSet) {
        return scheduler.isTaskSetFinished(taskSet);
    }

    @Override
    public Collection<Object> getResults(BaseTaskSet taskSet) {
        return scheduler.getTaskSetResults(taskSet);
    }

    @Override
    @GET
    @Path("/gettask/{client-id}")
    @Produces(MediaType.WILDCARD)
    public byte[] getTask(@PathParam("client-id") final String clientId) {
        final StopWatch stopWatch = StopWatchFactory.createOrGetLocal();
        stopWatch.start();
        final Ticket ticket = scheduler.getNextTicketFor(clientId);
        stopWatch.lap("server__broker-scheduler-next-ticket");
        final BaseTask<?, ?> task = ticket.getTask();
        try {
            final byte[] response = serializer.serializeTaskAndInput(task, ticket.getArguments());
            stopWatch.lap("server__broker-ticket-serialization");
            setTaskHeader(ticket, clientId);
            log.info("Broker sent to client #" + clientId + " task #" + task.getId() + " " + task.getClass().getSimpleName());
            return response;
        } catch (Exception e) {
            log.error("Broker can't serialize task: " + Pair.of(task.getClass().getSimpleName(), ticket.getArguments()), e);
            return null;
        }
    }

    @Override
    @POST
    @Path("/holdresult/{client-id}/{task-id}")
    @Consumes(MediaType.WILDCARD)
    public void holdResult(@PathParam("client-id") final String clientId,
                           @PathParam("task-id") final long taskId,
                           final byte[] data) {
        final StopWatch stopWatch = StopWatchFactory.createOrGetLocal();
        stopWatch.start();
        final Object result = serializer.deserializeResult(data);
        stopWatch.lap("server__broker-task-result-deserialization");
        log.info("TaskDescriptor #" + taskId + " finished with result " + result.toString());
        scheduler.taskFinished(clientId, taskId, result);
        stopWatch.lap("server__broker-scheduler-task-finished");
    }

    private void setTaskHeader(Ticket ticket, String clientId) {
        final BaseTask task = ticket.getTask();
        List<TaskEvent> taskEvents = new LinkedList<TaskEvent>();
        if (task instanceof BatchTask) {
            for (Ticket subTicket : (List<Ticket>) ticket.getArguments()) {
                final BaseTask subTask = subTicket.getTask();
                final TaskEvent taskEvent = TaskEvent.getDummyTaskEvent();
                taskEvent.setTaskId(subTask.getId() + "");
                taskEvent.setTaskType(subTask.getType());
                taskEvent.setOriginatorId(clientId);
                taskEvents.add(taskEvent);
            }
        } else {
            final TaskEvent taskEvent = TaskEvent.getDummyTaskEvent();
            taskEvent.setTaskId(task.getId() + "");
            taskEvent.setTaskType(task.getType());
            taskEvent.setOriginatorId(clientId);
            taskEvents.add(taskEvent);
        }
        final String encodedEvent = new String(serializer.encodeTaskEvents(taskEvents));
        servletResponse.setHeader(NetworkUtils.X_EVENT, encodedEvent);
    }


    public void setScheduler(Scheduler scheduler) {
        if (this.scheduler != null) {
            log.debug("Scheduler is disabled: {}", scheduler);
        }
        this.scheduler = scheduler;
        log.debug("Set scheduler: {}" , scheduler);
    }

    @Override
    public String toString() {
        return "BrokerImpl{" +
                "scheduler=" + scheduler +
                '}';
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        log.info("Set ApplicationContext: " + applicationContext);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (defaultTaskSetEmitter != null) {
            defaultTaskSetEmitter.start();
        }
    }
}
